package com.ibm.cps.storm;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;

import com.ibm.cps.message.MessageFields;
import com.ibm.cps.message.OptimizedParsedDataMessage;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.cps.message.RawDataMessage;
import com.ibm.cps.newmessage.BasicMetadata;
import com.ibm.cps.newmessage.JointMetadata;
import com.ibm.cps.newmessage.OptimizedMetadata;
import com.ibm.cps.newmessage.SerializableDataSourceMetadata;
import com.ibm.cps.processors.deletion.BFSSearch;
import com.ibm.cps.processors.deletion.PartialSearch;
import com.ibm.util.LocalConfig;
import com.ibm.util.TopologyComponentsId;
import com.ibm.util.TopologyMessageFieldIds;
import com.ibm.util.TopologyStreamIds;

public class
		CPSTopology {

	public void run(String tenantid) throws AlreadyAliveException, InvalidTopologyException {
		TopologyBuilder builder = new TopologyBuilder();
		builder.setSpout(TopologyComponentsId.MESSAGE_SPOUT_ID, new MessageSpout(),
				LocalConfig.STORM_MESSAGE_SPOUT_NUM);
		builder.setSpout(TopologyComponentsId.PROCESSOR_SPOUT_ID, new ProcessorSpout(),
				LocalConfig.STORM_PROCESSOR_SPOUT_NUM);
		builder.setSpout(TopologyComponentsId.DATASOURCE_SPOUT_ID, new DataSourceSpout(),
				LocalConfig.STORM_DATASOURCE_SPOUT_NUM);

		builder.setBolt(TopologyComponentsId.DATASOURCE_BOLT_ID, new DataSourceBolt(),
				LocalConfig.STORM_DATASOURCE_BOLT_NUM)
				.allGrouping(TopologyComponentsId.DATASOURCE_SPOUT_ID, TopologyStreamIds.DATASOURCE_SPOUT_STREAM)
				.allGrouping(TopologyComponentsId.DATASOURCE_SPOUT_ID, TopologyStreamIds.DATASOURCE_DELETE_SPOUT_STREAM)
				.fieldsGrouping(TopologyComponentsId.MESSAGE_SPOUT_ID, TopologyStreamIds.RAW_DATA_STREAM,
						new Fields(TopologyMessageFieldIds.TS_KEY));

		builder.setBolt(TopologyComponentsId.DISPATCH_BOLT_ID, new DispatchBolt(), LocalConfig.STORM_DISPATCH_BOLT_NUM)
				.allGrouping(TopologyComponentsId.PROCESSOR_SPOUT_ID, TopologyStreamIds.PROCESSOR_SPOUT_STREAM)
				.allGrouping(TopologyComponentsId.PROCESSOR_SPOUT_ID, TopologyStreamIds.PROCESSOR_DELETE_SPOUT_STREAM)
				.fieldsGrouping(TopologyComponentsId.DATASOURCE_BOLT_ID, TopologyStreamIds.PARSED_DATA_STREAM,
						new Fields(TopologyMessageFieldIds.TS_KEY))
				.fieldsGrouping(TopologyComponentsId.PROCESSOR_BOLT_ID, TopologyStreamIds.STREAM_TO_DISPATCH,
						new Fields(TopologyMessageFieldIds.TS_KEY));

		builder.setBolt(TopologyComponentsId.PROCESSOR_BOLT_ID, new ProcessorBolt(),
				LocalConfig.PROCESSOR_BOLT_TASK_NUM)
				.allGrouping(TopologyComponentsId.DISPATCH_BOLT_ID, TopologyStreamIds.METADATA_STREAM)
				.allGrouping(TopologyComponentsId.DISPATCH_BOLT_ID, TopologyStreamIds.METADATA_DELETE_STREAM)
				.fieldsGrouping(TopologyComponentsId.DISPATCH_BOLT_ID, TopologyStreamIds.DATAMESSAGE_STREAM,
						new Fields(TopologyMessageFieldIds.TS_KEY));

		// add config file path
		Config conf = buildConfigure(tenantid);
		if (LocalConfig.CPS_MODE.equals("STORM_LOCAL")) {
			// local cluster
			LocalCluster cluster = new LocalCluster();

			cluster.submitTopology("CPS-Topology-" + tenantid, conf, builder.createTopology());
		} else {
			// cluster
			conf.setNumWorkers(LocalConfig.WORKER_NUM);
			StormSubmitter.submitTopology("CPS-Topology-" + tenantid, conf, builder.createTopology());
		}
	}

	public static Config buildConfigure(String tenantid) {
		Config conf = new Config();
		conf.put("zookeeper", LocalConfig.ZOOKEEPER_HOST_PORT);
		conf.put(MessageFields.TENANTID, tenantid);
		conf.registerSerialization(SerializableDataSourceMetadata.class);
		conf.registerSerialization(RawDataMessage.class);
		conf.registerSerialization(ParsedDataMessage.class);
		conf.registerSerialization(OptimizedParsedDataMessage.class);
		conf.registerSerialization(BasicMetadata.class);
		conf.registerSerialization(OptimizedMetadata.class);
		conf.registerSerialization(JointMetadata.class);
		conf.registerSerialization(BFSSearch.class);
		conf.registerSerialization(PartialSearch.class);
		return conf;
	}
}
